package com.zyy.joseata.socket;

import com.alibaba.fastjson.JSONObject;
import com.zyy.joseata.bo.SocketBO;
import com.zyy.joseata.config.SeataConfigProperties;
import com.zyy.joseata.constant.TxConst;
import com.zyy.joseata.util.ThreadLocalUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;

@Component
@Slf4j
public class SocketClient {

    @Resource
    private SeataConfigProperties seataConfigProperties;

    private Bootstrap bootstrap;

    private ChannelFuture channelFuture;

    private String serverIp = "192.168.201.99"; //

    private int serverPort = 9188;

    @PostConstruct
    public void start(){
        bootstrap=new Bootstrap();
        bootstrap.group(new NioEventLoopGroup());
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                log.info("JoSeata连接成功, ServerIp:{}, port:{}",serverIp,serverPort);
                ChannelPipeline pipeline = nioSocketChannel.pipeline();
                ByteBuf delimiter = Unpooled.copiedBuffer("\n".getBytes());
                pipeline.addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                pipeline.addLast("decoder",new StringEncoder());
                pipeline.addLast("encoder",new StringEncoder());
                pipeline.addLast(new SocketReadHandler());
            }
        });

        if(seataConfigProperties != null) {
            serverIp = seataConfigProperties.getSeataServerIp();
            serverPort = seataConfigProperties.getSeataServerPort();
        }

        channelFuture=bootstrap.connect(new InetSocketAddress(serverIp,serverPort));
    }

    public void send(SocketBO socketData) {
        String dataStr = JSONObject.toJSONString(socketData)+"\n";
        // 创建要发送的数据，这里使用ByteBuf作为示例
        ByteBuf dataToSend = Unpooled.wrappedBuffer(dataStr.getBytes(StandardCharsets.UTF_8));
        try {
            channelFuture.sync().channel().writeAndFlush(dataToSend);
        } catch (InterruptedException e) {
            log.error("发送失败,",e);
        }
    }

    public void send(String eventType){
        SocketBO socketData = new SocketBO();
        socketData.setServiceName(seataConfigProperties.getServiceName());
        socketData.setEventType(eventType);
        socketData.setTxType(TxConst.TX_TYPE_TCC);
        socketData.setTxId(ThreadLocalUtil.getTxId());
        send(socketData);
    }

    public void send(String eventType,String message) throws InterruptedException{
        SocketBO socketData = new SocketBO();
        socketData.setTxType(TxConst.TX_TYPE_MQ);
        socketData.setServiceName(seataConfigProperties.getServiceName());
        socketData.setTxId(ThreadLocalUtil.getTxId());
        socketData.setEventType(eventType);
        socketData.setMqMsg(message);
        send(socketData);
    }
}
